package com.dd.keel.core.framework.server;


import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.buffer.SimpleBufferAllocator;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dd.keel.core.common.utils.CycleCounter;
import com.dd.keel.core.common.utils.FileUtil;
import com.dd.keel.core.common.utils.RandomUtil;
import com.dd.keel.core.framework.register.RegisterUtil;

/**
 * 服务器引擎
 * @author Kevin.XU
 *
 */
public class ServerEngine {

	final static private Logger LOGGER = LoggerFactory.getLogger(ServerEngine.class);
	
	final static CycleCounter receiverNumCycleCounter = new CycleCounter(0,10);
	
	/**接收服务器*/
	private IoAcceptor acceptor = null;
	private ExecutorService executor = null;	
	
	/**请求对象队列*/
	private ServerReqObjQueue serverReqObjQueue = null;
	
	/**工作者线程池*/
	private List<ServerWorker> serverWorkers = new ArrayList<ServerWorker>();
	
	/***参数***/
	private String engineName = "NA";         /**引擎名称*/
	private String listeningAddr = "0.0.0.0";
	private int listeningPort = 0;     /**侦听端口*/
	private int idleSeconds = 0;       /**最大空闲时间**/
	private int maxAllowClientNum = 0; /**最大允许客户端连接数量**/
	private int receiverCount = 0; /**接收者线程数目*/
	private int workerCount = 0;   /**工作者线程数目*/
	private int requestQueueMaxSize = 0;   /**请求队列中的最大对象数量*/
	private ProtocolCodecFactory protocolCodecFactory; /**协议处理器*/
	private ServerLogicHandler serverLogicHandler;     /**业务逻辑处理器*/
	private ServerHeartbeatHandler serverHeartbeatHandler;         /**心跳处理器*/
	private RegisterUtil registerUtil;
	private NodeConfig nodeConfig;
	
	/**
	 * 启动
	 */
	public void startup()
	{
		LOGGER.info("startup server engine ..." );
		setListeningAddr(nodeConfig.getHost());
		setListeningPort(nodeConfig.getPort());
		//startup
		if(getListeningPort()<=0 || getListeningPort()>65535)
		{
			LOGGER.error("listenning port is invalid , valid range is [1,65535]");
			return ;
		}
		if(getReceiverCount()<1)
		{
			LOGGER.error("receiver count value is invalid , valid range is [1,...]");
			return ;
		}
		if(getWorkerCount()<1)
		{
			LOGGER.error("worker thread count is invalid , valid range is [1,...]");
			return ;
		}
		if(getServerLogicHandler()==null)
		{
			LOGGER.error("server logic handler is null ");
			return ;
		}
		if(getProtocolCodecFactory()==null)
		{
			LOGGER.error("server protocol handler is null ");
			return ;
		}
		receiverNumCycleCounter.setMaxValue(getReceiverCount()-1);
		//
		serverReqObjQueue = new ServerReqObjQueue(getRequestQueueMaxSize());
		for(int workerIndex=0; workerIndex<getWorkerCount(); ++workerIndex)
		{
			serverWorkers.add( new ServerWorker(engineName,workerIndex,getServerLogicHandler(),serverReqObjQueue) );
		}
		//启动工作者线程池
		Iterator<ServerWorker> serverWorkerIter = serverWorkers.iterator();
		while(serverWorkerIter.hasNext())
		{
			serverWorkerIter.next().startup();
		}
		//启动接收消息服务器
		executor = Executors.newFixedThreadPool(getReceiverCount(),
			new ThreadFactory()
			{
				@Override
				public Thread newThread(Runnable r) 
				{
					Thread nwThread = new Thread(r,engineName+"-ServerReceiver-"+receiverNumCycleCounter.next());
					return nwThread;
				}
			}
		);
    	IoBuffer.setUseDirectBuffer(false);
    	IoBuffer.setAllocator(new SimpleBufferAllocator());
    	NioSocketAcceptor socketacceptor = new NioSocketAcceptor(Runtime.getRuntime().availableProcessors()+1);
        socketacceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(getProtocolCodecFactory()));
        socketacceptor.getFilterChain().addLast("threadPool", new ExecutorFilter(executor));
        ServerReceiverHandler serverReceiverHandler = new ServerReceiverHandler();
        serverReceiverHandler.setHeartbeatHandler(getServerHeartbeatHandler());
        serverReceiverHandler.setIdle_seconds(getIdleSeconds());
        serverReceiverHandler.setMax_connection_num(getMaxAllowClientNum());
        serverReceiverHandler.setServerReqObjQueue(serverReqObjQueue);
        socketacceptor.setHandler( serverReceiverHandler );
        socketacceptor.setDefaultLocalAddress(new InetSocketAddress(getListeningAddr(),getListeningPort()));
        socketacceptor.setReuseAddress(true);
        socketacceptor.setBacklog(10);
        acceptor = socketacceptor;
        //绑定端口
        try {
			acceptor.bind();
			LOGGER.info("{} is listening on port {}", getEngineName(), getListeningPort() );
		} catch (IOException e) {
			LOGGER.error("{} bind error on {}", getEngineName(), getListeningPort() );
			LOGGER.error(e.getMessage(), e);
			System.exit(1);
		}
        //register
        if(getRegisterUtil().registerServiceNode()){
        	LOGGER.info("register service successfully" );
        }else{
        	LOGGER.error("register service failed" );
        	System.exit(0);
        }
	}

	
	/**
	 * 关闭
	 */
	public void shutdown()
	{
		LOGGER.info("shutdown server engine ..." );
		getRegisterUtil().unregisterServiceNode();
		//关闭线程池，关闭顺序与启动顺序相反
		//关闭接收消息服务器
		if(acceptor!=null)
		{
			try{
				acceptor.dispose();
			}catch(Exception ex){
				LOGGER.error( ex.getMessage(), ex );
			}
		}
		if(executor!=null)
		{
			try{
				executor.shutdown();
			}catch(Exception ex){
				LOGGER.error( ex.getMessage(), ex );
			}
		}
		//关闭工作者线程池
		Iterator<ServerWorker> serverWorkerIter = serverWorkers.iterator();
		while(serverWorkerIter.hasNext())
		{
			serverWorkerIter.next().shutdown();
		}
		//清除线程池
		serverWorkers.clear();
		//清除队列内容
		serverReqObjQueue.clear();
	}
	
	public String getEngineName() {
		return engineName;
	}

	public void setEngineName(String engineName) {
		this.engineName = engineName;
	}

	public String getListeningAddr() {
		return listeningAddr;
	}

	public void setListeningAddr(String listeningAddr) {
		this.listeningAddr = listeningAddr;
	}


	public int getListeningPort() {
		return listeningPort;
	}

	public void setListeningPort(int listeningPort) {
		if(listeningPort>0) this.listeningPort = listeningPort;
	}

	public int getIdleSeconds() {
		return idleSeconds;
	}

	public void setIdleSeconds(int idleSeconds) {
		if(idleSeconds>0) this.idleSeconds = idleSeconds;
	}

	public int getReceiverCount() {
		return receiverCount;
	}

	public void setReceiverCount(int receiverCount) {
		if(receiverCount>0) this.receiverCount = receiverCount;
	}

	public int getWorkerCount() {
		return workerCount;
	}

	public void setWorkerCount(int workerCount) {
		if(workerCount>0) this.workerCount = workerCount;
	}

	public int getRequestQueueMaxSize() {
		return requestQueueMaxSize;
	}

	public void setRequestQueueMaxSize(int requestQueueMaxSize) {
		if(requestQueueMaxSize>0) this.requestQueueMaxSize = requestQueueMaxSize;
	}

	public ServerLogicHandler getServerLogicHandler() {
		return serverLogicHandler;
	}

	public void setServerLogicHandler(ServerLogicHandler serverLogicHandler) {
		this.serverLogicHandler = serverLogicHandler;
	}

	public ProtocolCodecFactory getProtocolCodecFactory() {
		return protocolCodecFactory;
	}
	
	public void setProtocolCodecFactory(ProtocolCodecFactory protocolCodecFactory) {
		this.protocolCodecFactory = protocolCodecFactory;
	}

	public ServerHeartbeatHandler getServerHeartbeatHandler() {
		return serverHeartbeatHandler;
	}

	public void setServerHeartbeatHandler(
			ServerHeartbeatHandler serverHeartbeatHandler) {
		this.serverHeartbeatHandler = serverHeartbeatHandler;
	}

	public int getMaxAllowClientNum() {
		return maxAllowClientNum;
	}

	public void setMaxAllowClientNum(int maxAllowClientNum) {
		if(maxAllowClientNum>0) this.maxAllowClientNum = maxAllowClientNum;
	}

	public RegisterUtil getRegisterUtil() {
		return registerUtil;
	}

	public void setRegisterUtil(RegisterUtil registerUtil) {
		this.registerUtil = registerUtil;
	}

	public NodeConfig getNodeConfig() {
		return nodeConfig;
	}

	public void setNodeConfig(NodeConfig nodeConfig) {
		this.nodeConfig = nodeConfig;
	}
	
	
	
}
